﻿using System.Globalization;
using RockBus.DataContracts;
using RockBus.Logging.Interfaces;
using RockBus.MessageRepository.Interfaces;
using System;
using System.Collections.Generic;
using System.Configuration;
using System.Data;
using System.Data.SqlClient;
using System.Reflection;
using System.Text;
using System.Threading.Tasks;
using System.Transactions;

namespace RockBus.MessageRepository.SqlServer
{
    public class MessageRepository : IMessageRepository
    {
        public MessageRepository(ILogger logger, IDbMessageRepositoryConfiguration configuration)
        {
            this.Logger = logger;
            this.Configuration = configuration;
        }

        public ILogger Logger { get; set; }

        public IDbMessageRepositoryConfiguration Configuration { get; set; }

        public long CreateMessageBuffer(
            string bufferId,
            long messageValidityDurationSecs,
            long maxMessageCount)
        {
            try
            {
                using (SqlConnection conn = new SqlConnection(this.Configuration.ConnectionString))
                {
                    conn.Open();

                    SqlCommand sc = new SqlCommand("MessageBuffer_InsertUpdate", conn);
                    sc.Parameters.Add("@bufferId", SqlDbType.NVarChar).Value = bufferId;
                    sc.Parameters.Add("@messageValidityDurationSecs", SqlDbType.BigInt).Value = messageValidityDurationSecs;
                    sc.Parameters.Add("@maxMessageCount", SqlDbType.BigInt).Value = maxMessageCount;

                    // Output parameters
                    SqlParameter returnCodeParameter = new SqlParameter("@returnCode", 0);
                    returnCodeParameter.SqlDbType = System.Data.SqlDbType.Int;
                    returnCodeParameter.Direction = System.Data.ParameterDirection.Output;
                    sc.Parameters.Add(returnCodeParameter);

                    SqlParameter bufferDbIdParameter = new SqlParameter("@bufferDbId", 0);
                    bufferDbIdParameter.SqlDbType = System.Data.SqlDbType.BigInt;
                    bufferDbIdParameter.Direction = System.Data.ParameterDirection.Output;
                    sc.Parameters.Add(bufferDbIdParameter);

                    sc.CommandType = CommandType.StoredProcedure;
                    sc.ExecuteNonQuery();

                    return (long)bufferDbIdParameter.Value;
                }
            }
            catch (Exception exc)
            {
                this.Logger.Log(MethodBase.GetCurrentMethod(), exc);
                throw;
            }
        }

        public long CreateMessage(
            string correlationId,
            string messageVersion,
            string action,
            DateTime receiveTime,
            string receiveAddress)
        {
            try
            {
                using (SqlConnection conn = new SqlConnection(this.Configuration.ConnectionString))
                {
                    conn.Open();

                    SqlCommand sc = new SqlCommand("CreateMessage", conn);
                    sc.Parameters.Add("@correlationId", SqlDbType.NVarChar).Value = correlationId;
                    sc.Parameters.Add("@messageVersion", SqlDbType.NVarChar).Value = messageVersion;
                    sc.Parameters.Add("@action", SqlDbType.NVarChar).Value = action;
                    sc.Parameters.Add("@receiveTime", SqlDbType.DateTime2).Value = receiveTime;
                    sc.Parameters.Add("@receiveAddress", SqlDbType.NVarChar).Value = receiveAddress;

                    // Output parameters
                    SqlParameter returnCodeParameter = new SqlParameter("@returnCode", 0);
                    returnCodeParameter.SqlDbType = System.Data.SqlDbType.Int;
                    returnCodeParameter.Direction = System.Data.ParameterDirection.Output;
                    sc.Parameters.Add(returnCodeParameter);

                    SqlParameter messageIdParameter = new SqlParameter("@messageId", 0);
                    messageIdParameter.SqlDbType = System.Data.SqlDbType.BigInt;
                    messageIdParameter.Direction = System.Data.ParameterDirection.Output;
                    sc.Parameters.Add(messageIdParameter);

                    sc.CommandType = CommandType.StoredProcedure;
                    sc.ExecuteNonQuery();

                    return (long)messageIdParameter.Value;
                }
            }
            catch (Exception exc)
            {
                this.Logger.Log(MethodBase.GetCurrentMethod(), exc);
                throw;
            }
        }

        public long CreateMessageFragment(
            long messageId,
            int partIndex,
            byte[] body,
            int size)
        {
            try
            {
                using (SqlConnection conn = new SqlConnection(this.Configuration.ConnectionString))
                {
                    conn.Open();

                    SqlCommand scPart = new SqlCommand("CreateMessageFragment", conn);
                    scPart.Parameters.Add("@messageId", SqlDbType.BigInt).Value = messageId;
                    scPart.Parameters.Add("@partIndex", SqlDbType.Int).Value = partIndex;
                    scPart.Parameters.Add("@size", SqlDbType.Int).Value = size;
                    scPart.Parameters.Add("@body", SqlDbType.VarBinary, size).Value = body;

                    // Output parameters
                    SqlParameter returnCodeParameter = new SqlParameter("@returnCode", 0);
                    returnCodeParameter.SqlDbType = System.Data.SqlDbType.Int;
                    returnCodeParameter.Direction = System.Data.ParameterDirection.Output;
                    scPart.Parameters.Add(returnCodeParameter);

                    SqlParameter messageFragmentIdParameter = new SqlParameter("@messageFragmentId", 0);
                    messageFragmentIdParameter.SqlDbType = System.Data.SqlDbType.BigInt;
                    messageFragmentIdParameter.Direction = System.Data.ParameterDirection.Output;
                    scPart.Parameters.Add(messageFragmentIdParameter);

                    scPart.CommandType = CommandType.StoredProcedure;
                    scPart.ExecuteNonQuery();

                    return (long)messageFragmentIdParameter.Value;
                }
            }
            catch (Exception exc)
            {
                this.Logger.Log(MethodBase.GetCurrentMethod(), exc);
                throw;
            }
        }

        public long EnqueueEntry(
            long messageId,
            byte[] routingHeaderByteArray,
            int priority,
            bool orderedDelivery,
            string receiverMachineName,
            int receiverProcessId,
            int receiverAppDomainId,
            long? messageValidityDurationSecs,
            long deliveryAttemptIntervalSecs,
            int maxDeliveryAttemptCount,
            bool isResponse)
        {
            try
            {
                using (SqlConnection conn = new SqlConnection(this.Configuration.ConnectionString))
                {
                    SqlCommand sc = new SqlCommand("EnqueueEntry", conn);
                    sc.CommandType = CommandType.StoredProcedure;

                    sc.Parameters.Add("@messageId", SqlDbType.BigInt).Value = messageId;
                    sc.Parameters.Add("@routingHeader", SqlDbType.VarBinary).Value = routingHeaderByteArray;
                    sc.Parameters.Add("@priority", SqlDbType.Int).Value = priority;
                    sc.Parameters.Add("@orderedDelivery", SqlDbType.Int).Value = orderedDelivery;
                    sc.Parameters.Add("@receiverMachineName", SqlDbType.VarChar).Value = receiverMachineName;
                    sc.Parameters.Add("@receiverProcessId", SqlDbType.Int).Value = receiverProcessId;
                    sc.Parameters.Add("@receiverAppDomainId", SqlDbType.Int).Value = receiverAppDomainId;
                    if (messageValidityDurationSecs.HasValue)
                    {
                        sc.Parameters.Add("@messageValidityDurationSecs", SqlDbType.BigInt).Value = messageValidityDurationSecs.Value;
                    }
                    else
                    {
                        sc.Parameters.Add("@messageValidityDurationSecs", SqlDbType.BigInt).Value = DBNull.Value;
                    }
                    sc.Parameters.Add("@deliveryAttemptIntervalSecs", SqlDbType.BigInt).Value = deliveryAttemptIntervalSecs;
                    sc.Parameters.Add("@maxDeliveryAttemptCount", SqlDbType.Int).Value = maxDeliveryAttemptCount;
                    sc.Parameters.Add("@isResponse", SqlDbType.Bit).Value = isResponse;

                    // Output parameters
                    SqlParameter returnCodeParameter = new SqlParameter("@returnCode", 0);
                    returnCodeParameter.SqlDbType = System.Data.SqlDbType.Int;
                    returnCodeParameter.Direction = System.Data.ParameterDirection.Output;
                    sc.Parameters.Add(returnCodeParameter);

                    SqlParameter queueEntryIdParameter = new SqlParameter("@queueEntryId", 0);
                    queueEntryIdParameter.SqlDbType = System.Data.SqlDbType.BigInt;
                    queueEntryIdParameter.Direction = System.Data.ParameterDirection.Output;
                    sc.Parameters.Add(queueEntryIdParameter);

                    conn.Open();
                    sc.ExecuteNonQuery();

                    return (long)queueEntryIdParameter.Value;
                }
            }
            catch (Exception exc)
            {
                this.Logger.Log(MethodBase.GetCurrentMethod(), exc);
                throw;
            }
        }

        public void CompleteMessage(
            long messageId,
            int partCount,
            int size)
        {
            try
            {
                using (SqlConnection conn = new SqlConnection(this.Configuration.ConnectionString))
                {
                    Logger.Log(MethodBase.GetCurrentMethod(), LogLevelEnum.Debug, Transaction.Current);

                    SqlCommand sc = new SqlCommand("CompleteMessage", conn);
                    sc.CommandType = CommandType.StoredProcedure;

                    sc.Parameters.Add("@messageId", SqlDbType.BigInt).Value = messageId;
                    sc.Parameters.Add("@partCount", SqlDbType.Int).Value = partCount;
                    sc.Parameters.Add("@size", SqlDbType.Int).Value = size;

                    // Output parameters
                    SqlParameter returnCodeParameter = new SqlParameter("@returnCode", 0);
                    returnCodeParameter.SqlDbType = System.Data.SqlDbType.Int;
                    returnCodeParameter.Direction = System.Data.ParameterDirection.Output;
                    sc.Parameters.Add(returnCodeParameter);

                    conn.Open();
                    sc.ExecuteNonQuery();
                }
            }
            catch (Exception exc)
            {
                this.Logger.Log(MethodBase.GetCurrentMethod(), exc);
                throw;
            }
        }

        public Task<DequeueResult> StartDequeueEnties(
            int maxEntryCount,
            string machineName,
            int processId,
            int appDomainId,
            string processorId)
        {
            try
            {
                SqlConnection conn = new SqlConnection(this.Configuration.ConnectionString);
                SqlCommand sc = new SqlCommand("DequeueEntries", conn);
                sc.CommandType = CommandType.StoredProcedure;

                sc.Parameters.Add("@count", SqlDbType.Int).Value = maxEntryCount;
                sc.Parameters.Add("@machineName", SqlDbType.NVarChar).Value = machineName;
                sc.Parameters.Add("@processId", SqlDbType.Int).Value = processId;
                sc.Parameters.Add("@appDomainId", SqlDbType.Int).Value = appDomainId;
                sc.Parameters.Add("@processorId", SqlDbType.NVarChar).Value = processorId;

                // Output parameters
                SqlParameter returnCodeParameter = new SqlParameter("@returnCode", 0);
                returnCodeParameter.SqlDbType = System.Data.SqlDbType.Int;
                returnCodeParameter.Direction = System.Data.ParameterDirection.Output;
                sc.Parameters.Add(returnCodeParameter);

                SqlParameter batchIdParameter = new SqlParameter("@batchId", 0);
                batchIdParameter.SqlDbType = System.Data.SqlDbType.UniqueIdentifier;
                batchIdParameter.Direction = System.Data.ParameterDirection.Output;
                sc.Parameters.Add(batchIdParameter);

                conn.Open();
                return Task<DequeueResult>.Factory.FromAsync(
                    sc.BeginExecuteReader,
                    this.EndDequeueEntries,
                    sc);
            }
            catch (Exception exc)
            {
                this.Logger.Log(MethodBase.GetCurrentMethod(), exc);
                throw;
            }
        }

        public DequeueResult EndDequeueEntries(IAsyncResult asyncResult)
        {
            SqlCommand sc = null;
            try
            {
                sc = asyncResult.AsyncState as SqlCommand;
                List<QueueEntry> queueEntries = new List<QueueEntry>();
                using (SqlDataReader reader = sc.EndExecuteReader(asyncResult))
                {
                    if (reader.RecordsAffected == 0)
                    {
                        this.Logger.LogDebug(MethodBase.GetCurrentMethod(), "No entries to dequeue.");
                        return null;
                    }

                    // Map reader to DequeueResult
                    while (reader.Read())
                    {
                        QueueEntry qe = new QueueEntry()
                        {
                            Id = (long)reader["Id"],
                            MessageId = (long)reader["MessageId"],
                            BatchId = (Guid)reader["BatchId"],
                            Priority = (int)reader["Priority"],
                            OrderedDelivery = (bool)reader["OrderedDelivery"],
                            ReceiverMachineName = (string)reader["ReceiverMachineName"],
                            ReceiverProcessId = (int)reader["ReceiverProcessId"],
                            ReceiverAppDomainId = (int)reader["ReceiverAppDomainId"],
                            MachineName = (string)reader["MachineName"],
                            ProcessId = (int)reader["ProcessId"],
                            ProcessorId = (string)reader["ProcessorId"],
                            EnqueueTime = (DateTime)reader["EnqueueTime"],
                            DequeueTime = (DateTime)reader["DequeueTime"],
                            DequeueAttemptCount = (int)reader["DequeueAttemptCount"],
                            PartCount = (int)reader["PartCount"],
                            Size = (int)reader["Size"],
                            CorrelationId = (string)reader["CorrelationId"],
                            MessageVersion = (string)reader["MessageVersion"],
                            Action = (string)reader["Action"],
                            ReceiveTime = (DateTime)reader["ReceiveTime"],
                            ReceiveAddress = (string)reader["ReceiveAddress"],
                            CreateTime = (DateTime)reader["CreateTime"],
                        };

                        if ((null != reader["ExpirationTime"]) && (DBNull.Value != reader["ExpirationTime"]))
                        {
                            qe.ExpirationTime = (DateTime?)reader["ExpirationTime"];
                        }

                        byte[] routingHeaderByteArray = (byte[])reader["RoutingHeader"];
                        qe.RoutingHeader = RoutingHeader.CreateFromByteArray(routingHeaderByteArray);

                        queueEntries.Add(qe);
                    }
                }

                // Retrieve output parameters
                //  Output parameters can only be retrieved when the Sqlreader is read entirely
                int? returnCode = (int?)sc.Parameters["@returnCode"].Value;
                if ((null == returnCode) || (returnCode.Value != 0))
                {
                    this.Logger.LogError(MethodBase.GetCurrentMethod(), "Return code was {0}.", returnCode);
                    return null;
                }

                Guid? batchId = (Guid?)sc.Parameters["@batchId"].Value;
                if ((null == batchId) || (batchId.Value == Guid.Empty))
                {
                    this.Logger.LogError(MethodBase.GetCurrentMethod(), "BatchId was NULL.");
                    return null;
                }

                DequeueResult result = new DequeueResult(batchId.Value, queueEntries);
                return result;
            }
            catch (Exception exc)
            {
                this.Logger.Log(MethodBase.GetCurrentMethod(), exc);
                throw;
            }
            finally
            {
                sc.Connection.Close();
            }
        }

        public void BufferMessage(string bufferId, long messageId)
        {
            try
            {
                using (SqlConnection conn = new SqlConnection(this.Configuration.ConnectionString))
                {
                    SqlCommand sc = new SqlCommand("BufferMessage", conn);
                    sc.CommandType = CommandType.StoredProcedure;

                    sc.Parameters.Add("@bufferId", SqlDbType.NVarChar).Value = bufferId;
                    sc.Parameters.Add("@messageId", SqlDbType.BigInt).Value = messageId;

                    // Output parameters
                    SqlParameter returnCodeParameter = new SqlParameter("@returnCode", 0);
                    returnCodeParameter.SqlDbType = System.Data.SqlDbType.Int;
                    returnCodeParameter.Direction = System.Data.ParameterDirection.Output;
                    sc.Parameters.Add(returnCodeParameter);

                    conn.Open();
                    sc.ExecuteNonQuery();
                }
            }
            catch (Exception exc)
            {
                this.Logger.Log(MethodBase.GetCurrentMethod(), exc);
                throw;
            }
        }

        public byte[] GetMessageFragment(long messageId, int partIndex)
        {
            try
            {
                using (SqlConnection conn = new SqlConnection(this.Configuration.ConnectionString))
                {
                    SqlCommand sc = new SqlCommand("GetMessageFragment", conn);
                    sc.CommandType = CommandType.StoredProcedure;

                    sc.Parameters.Add("@messageId", SqlDbType.BigInt).Value = messageId;
                    sc.Parameters.Add("@partIndex", SqlDbType.Int).Value = partIndex;

                    conn.Open();
                    using (SqlDataReader reader = sc.ExecuteReader())
                    {
                        if (!reader.Read())
                        {
                            this.Logger.LogError(MethodBase.GetCurrentMethod(), "MessageFragment could not be read. MessageId: {0}; PartIndex: {1}.", messageId.ToString(CultureInfo.CurrentCulture), partIndex.ToString(CultureInfo.CurrentCulture));
                            return null;
                        }

                        return (byte[])reader["Body"];
                    }
                }
            }
            catch (Exception exc)
            {
                this.Logger.Log(MethodBase.GetCurrentMethod(), exc);
                throw;
            }
        }

        public void CompleteEntry(long queueEntryId)
        {
            try
            {
                using (SqlConnection conn = new SqlConnection(this.Configuration.ConnectionString))
                {
                    Logger.Log(MethodBase.GetCurrentMethod(), LogLevelEnum.Debug, Transaction.Current);

                    SqlCommand sc = new SqlCommand("CompleteEntry", conn);
                    sc.CommandType = CommandType.StoredProcedure;

                    sc.Parameters.Add("@queueEntryId", SqlDbType.BigInt).Value = queueEntryId;

                    // Output parameters
                    SqlParameter returnCodeParameter = new SqlParameter("@returnCode", 0);
                    returnCodeParameter.SqlDbType = System.Data.SqlDbType.Int;
                    returnCodeParameter.Direction = System.Data.ParameterDirection.Output;
                    sc.Parameters.Add(returnCodeParameter);

                    conn.Open();
                    sc.ExecuteNonQuery();
                }
            }
            catch (Exception exc)
            {
                this.Logger.Log(MethodBase.GetCurrentMethod(), exc);
                throw;
            }
        }

        public void RollbackEntry(long queueEntryId, Exception exception)
        {
            try
            {
                using (SqlConnection conn = new SqlConnection(this.Configuration.ConnectionString))
                {
                    SqlCommand sc = new SqlCommand("RollbackEntry", conn);
                    sc.CommandType = CommandType.StoredProcedure;

                    sc.Parameters.Add("@queueEntryId", SqlDbType.BigInt).Value = queueEntryId;
                    sc.Parameters.Add("@errorMsg", SqlDbType.NVarChar).Value = exception.ToString();

                    // Output parameters
                    SqlParameter returnCodeParameter = new SqlParameter("@returnCode", 0);
                    returnCodeParameter.SqlDbType = System.Data.SqlDbType.Int;
                    returnCodeParameter.Direction = System.Data.ParameterDirection.Output;
                    sc.Parameters.Add(returnCodeParameter);

                    conn.Open();
                    sc.ExecuteNonQuery();
                }
            }
            catch (Exception exc)
            {
                this.Logger.Log(MethodBase.GetCurrentMethod(), exc);
                throw;
            }
        }

        public void FaultEntry(long queueEntryId, Exception exception)
        {
            try
            {
                using (SqlConnection conn = new SqlConnection(this.Configuration.ConnectionString))
                {
                    SqlCommand sc = new SqlCommand("FaultEntry", conn);
                    sc.CommandType = CommandType.StoredProcedure;

                    sc.Parameters.Add("@queueEntryId", SqlDbType.BigInt).Value = queueEntryId;
                    sc.Parameters.Add("@errorMsg", SqlDbType.NVarChar).Value = exception.ToString();

                    // Output parameters
                    SqlParameter returnCodeParameter = new SqlParameter("@returnCode", 0);
                    returnCodeParameter.SqlDbType = System.Data.SqlDbType.Int;
                    returnCodeParameter.Direction = System.Data.ParameterDirection.Output;
                    sc.Parameters.Add(returnCodeParameter);

                    conn.Open();
                    sc.ExecuteNonQuery();
                }
            }
            catch (Exception exc)
            {
                this.Logger.Log(MethodBase.GetCurrentMethod(), exc);
                throw;
            }
        }

        public void CreateErrorMessage(
            long messageId,
            byte[] routingHeaderByteArray,
            string machineName,
            int processId,
            int appDomainId,
            int deliveryAttemptCount,
            Exception exception)
        {
            try
            {
                using (SqlConnection conn = new SqlConnection(this.Configuration.ConnectionString))
                {
                    SqlCommand sc = new SqlCommand("CreateErrorMessage", conn);
                    sc.CommandType = CommandType.StoredProcedure;

                    sc.Parameters.Add("@messageId", SqlDbType.BigInt).Value = messageId;
                    sc.Parameters.Add("@routingHeader", SqlDbType.VarBinary).Value = routingHeaderByteArray;
                    sc.Parameters.Add("@machineName", SqlDbType.VarChar).Value = machineName;
                    sc.Parameters.Add("@processId", SqlDbType.Int).Value = processId;
                    sc.Parameters.Add("@appDomainId", SqlDbType.Int).Value = appDomainId;
                    sc.Parameters.Add("@deliveryAttemptCount", SqlDbType.Int).Value = deliveryAttemptCount;
                    sc.Parameters.Add("@errorMsg", SqlDbType.NVarChar).Value = exception.ToString();

                    conn.Open();
                    sc.ExecuteNonQuery();
                }
            }
            catch (Exception exc)
            {
                this.Logger.Log(MethodBase.GetCurrentMethod(), exc);
                throw;
            }
        }

        private string ToDelimitedString<T>(IEnumerable<T> items, char seperator = ',')
        {
            StringBuilder delimitedString = new StringBuilder();
            foreach (var item in items)
            {
                delimitedString.Append(item.ToString());
                delimitedString.Append(seperator);
            }
            return delimitedString.ToString();
        }

        private IEnumerable<string> FromDelimitedString(string delimitedString, char seperator = ',')
        {
            string[] fragments = delimitedString.Split(seperator);
            foreach (var fragment in fragments)
            {
                yield return fragment;
            }
        }
    }
}